package com.zhangpan.realtime.common.util;

import com.alibaba.fastjson.JSONObject;
import com.google.common.base.CaseFormat;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.beanutils.BeanUtils;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.Cell;
import org.apache.hadoop.hbase.CellUtil;
import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.hbase.client.*;
import org.apache.hadoop.hbase.util.Bytes;

import java.io.IOException;
import java.util.List;

@Slf4j
public class HBaseUtil {

    public static Connection getHBaseConnection() throws IOException {
        Configuration conf = new Configuration();
        conf.set("hbase.zookeeper.quorum","hdp131");
        conf.set("hbase.zookeeper.property.clientPort","2181");
        return ConnectionFactory.createConnection(conf);
    }

    public static void closeHBaseConn(Connection hbaseConn) throws IOException {
        if (hbaseConn != null && !hbaseConn.isClosed()) {
            hbaseConn.close();
        }
    }

    public static void createHBaseTable(Connection hbaseConn,String nameSpace,String table,String family) throws IOException {

        Admin admin = hbaseConn.getAdmin();
        TableName tableName = TableName.valueOf(nameSpace,table);
        // 判断要建的表是否存在
        if (admin.tableExists(tableName)) {
            return;
        }

        // 列族描述器
        ColumnFamilyDescriptor cfDesc = ColumnFamilyDescriptorBuilder.of(family);

        TableDescriptor desc = TableDescriptorBuilder.newBuilder(tableName)
                .setColumnFamily(cfDesc)
                .build();
        admin.createTable(desc);
        admin.close();
        log.info("nameSpace" + " " + table + " 建表成功");
    }

    public static void dropHBaseTable(Connection hbaseConn,String nameSpace,String table) throws IOException {
        Admin admin = hbaseConn.getAdmin();
        TableName tableName = TableName.valueOf(nameSpace,table);
        if (admin.tableExists(tableName)) {
            admin.disableTable(tableName);
            admin.deleteTable(tableName);
        }
        admin.close();
        log.info(nameSpace + " " + table + " 删除成功");
    }

    public static void putRow(Connection con, String nameSpace, String table, String rowKey, String family, JSONObject data) throws IOException {

        //1. 获取table对象
        TableName tableName = TableName.valueOf(nameSpace,table);
        Table t = con.getTable(tableName);

        //2. 创建put对象
        Put put = new Put(Bytes.toBytes(rowKey));

        //3. 把每列放入到put 对象
        for(String key: data.keySet()) {
            String value = data.getString(key);
            if (value != null) {
                put.addColumn(Bytes.toBytes(family),Bytes.toBytes(key),Bytes.toBytes(value));
            }
        }

        //4. 想table对象中put 数据
        t.put(put);

        t.close();
    }

    public static void delRow(Connection con,String namespace,String table,String rowKey) throws IOException {
        TableName tableName = TableName.valueOf(namespace,table);
        Table t = con.getTable(tableName);

        Delete delete = new Delete(Bytes.toBytes(rowKey));
        // 删除整行
        t.delete(delete);
        t.close();
    }

    public static <T> T getRow(Connection hbaseConn,String nameSpace,String table,String rowKey,Class<T> tClass,boolean... isUnderlineToCamel) {

        boolean defaultIsUtoC = false; // 默认不执行下划线转驼峰
        if (isUnderlineToCamel.length > 0) {
            defaultIsUtoC = isUnderlineToCamel[0];
        }
        try(Table Table = hbaseConn.getTable(TableName.valueOf(nameSpace,table))) {
            Get get = new Get(Bytes.toBytes(rowKey));
            Result result = Table.get(get);
            //3. 将查询的数据封装到一个jsonObject 中
            //4.1 一行中的列全部解析出来
            List<Cell> cellList = result.listCells();
            T t = tClass.newInstance();
            for (Cell cell: cellList) {

                // 取出每列的列名
                String key = Bytes.toString(CellUtil.cloneQualifier(cell));
                if (defaultIsUtoC) {
                    key = CaseFormat.LOWER_UNDERSCORE.to(CaseFormat.LOWER_CAMEL,key);
                }
                String value = Bytes.toString(CellUtil.cloneValue(cell));

                BeanUtils.setProperty(t,key,value);
            }

            return t;
        }catch (Exception e) {
            throw new RuntimeException();
        }

    }

    /**
     * 异步从hbase 读取维度数据
     * @param hBaseAsyncConn
     * @param namespace
     * @param tableName
     * @param rowKey
     * @return
     */
    public static JSONObject readDimAsync(AsyncConnection hBaseAsyncConn,String namespace,String tableName,String rowKey) {

        AsyncTable<AdvancedScanResultConsumer> asyncTable = hBaseAsyncConn.getTable(TableName.valueOf(namespace,tableName));
        Get get = new Get(Bytes.toBytes(rowKey));
        try {
            // 获取result
            Result result = asyncTable.get(get).get();
            List<Cell> cellList = result.listCells();
            JSONObject dim = new JSONObject();
            for (Cell cell : cellList) {
                // 取出每列的列名
                String key = Bytes.toString(CellUtil.cloneQualifier(cell));
                String value = Bytes.toString(CellUtil.cloneValue(cell));
                dim.put(key,value);

            }

            return dim;
        }catch (Exception e) {

            e.printStackTrace();
            throw new RuntimeException(e);
        }


    }
}
